package com.test.reactor.service;

import java.time.Duration;
import java.util.List;
import java.util.Optional;
import java.util.logging.Level;

import org.apache.logging.log4j.LogManager;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;

import com.test.reactor.config.GlobalException;
import com.test.reactor.control.ClientController;

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;

@Service
@Slf4j
public class WebClientService {
	@Autowired
	TransactionMQProducer tproducer;
	@Autowired
	DefaultMQProducer producer;
	
	
	public Mono<String> clent(){
		String uri = "http://api.map.baidu.com/place/v2/search?ak=123";
		WebClient client = WebClient.create();
		Mono<String> ret = client.get()
				.uri(uri)
				.accept(MediaType.APPLICATION_JSON).exchange().doOnError(e -> {
					System.out.println(e);
				})
//				.timeout(Duration.ofMillis(1l))
				.flatMap(response -> response.bodyToMono(String.class))
				.log(ClientController.class.getName(), Level.ALL)
				;
		ret.subscribe(msg -> {
			log.info(msg);
			try {
				producer.send(new Message("topic2019","test", msg.getBytes()));
			} catch (Exception e) {
				log.error("", e);
			}
		});
		return ret;
	} 
	
	public Mono<String> transaction(){
		String uri = "http://api.map.baidu.com/place/v2/search?ak=123";
		WebClient client = WebClient.create();
		Mono<String> ret = client.get()
				.uri(uri)
				.accept(MediaType.APPLICATION_JSON).exchange()
				.flatMap(response -> response.bodyToMono(String.class))
				.map(msg -> {
					log.info(msg);
					try {
						Message m = new Message("topic2019","test", msg.getBytes());
						TransactionSendResult sendMessageInTransaction = tproducer.sendMessageInTransaction(
								m, null);
						log.info("send--" + sendMessageInTransaction.getMsgId());
					} catch (Exception e) {
						log.error("", e);
						return "{\"path\":\"/transaction\",\"status\":\"500\""
								+ ",\"error\":\"eeeeeeeeeeeeeeeee\",\"message\":\""+ e.getMessage()+"\"}";
					}
					return msg;
				})
//				.timeout(Duration.ofMillis(100000l), timeOut(uri))
				.log(ClientController.class.getName(), Level.ALL)
				;
		return ret;
	} 
	public Mono<String> orderMsg() throws MQClientException, RemotingException, MQBrokerException, InterruptedException{
		for (int i = 0; i < 10; i++) {
			Message m = new Message("topic2019","test", ("order"+i).getBytes());
			SendResult send = producer.send(m
//					, new MessageQueueSelector() {
//						@Override
//						public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
//							log.info("MessageQueue size--" + mqs.size());
//							Long id = (Long) arg;
//	                        long index = id % mqs.size();
//	                        return mqs.get((int)index);
//						}
//					}, 3l
				);
			log.info("send--" + send.getMsgId());
		}
		return Mono.just("{\"status\":200}");
	} 
	
	public Mono<String> timeOut(String msg){
		throw new GlobalException(HttpStatus.SERVICE_UNAVAILABLE, msg + " timeOut");
//		return Mono.just("{\"status\":\"500\",\"error\":\"timeout\"}");
	}
}
